/**
 * chenxitech.cn Inc.
 * Copyright (c) 2017-2017 All Rights Reserved.
 */
package tbdp.tool;

import com.google.gson.reflect.TypeToken;

import org.apache.commons.io.IOUtils;

import java.io.IOException;
import java.io.InputStream;
import java.util.ArrayList;
import java.util.List;

import tbdp.tool.common.ColumnDesc;
import tbdp.tool.common.JobConstants;
import tbdp.tool.common.StockInfo;
import tbdp.tool.config.ReaderConfig;
import tbdp.tool.context.JobContext;
import tbdp.tool.spi.ListRecordIterator;
import tbdp.tool.spi.Reader;
import tbdp.tool.spi.RecordIterator;
import tbdp.tool.util.GsonUtils;
import tbdp.tool.util.ResourceUtil;

/**
 * 从hbase 里面获取数据
 *
 * @author xbkaishui
 * @version $Id: HbaseReader.java, v 0.1 2017-06-17 下午3:06 xbkaishui Exp $$
 */
public class HbaseReader extends Reader {

    @Override
    public RecordIterator read(JobContext context) {
        HbaseRecordIterator recordIterator = null;
        ReaderConfig readerConfig = context.getJobConfig().getReader();
        List<StockInfo> stockInfos = (List<StockInfo>) context.getValue(JobConstants.STOCK_LIST);
        logger.info("stockInfos " + stockInfos);
        //如果为空不处理
        if (stockInfos == null || stockInfos.isEmpty()) {
            return new ListRecordIterator(new ArrayList());
        }
        try {
            String columnSchema = readerConfig.getConfigPropAsString(JobConstants.COLUMN_SCHEMA_PATH);
            List<ColumnDesc> cols = getColumnDesc(columnSchema);
            //set to context
            context.put(JobConstants.COLUMN_SCHEMA, cols.toArray(new ColumnDesc[]{}));
            recordIterator = new HbaseRecordIterator(context, cols);
            return recordIterator;
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    private List<ColumnDesc> getColumnDesc(String schema) throws IOException {
        InputStream ins = ResourceUtil.getInputStream(schema);
        String content = IOUtils.toString(ins, "UTF-8");
        return GsonUtils.fromJson(content, new TypeToken<List<ColumnDesc>>() {
        }.getType());
    }

    @Override
    public String getType() {
        return "hbase";
    }
}
